package com.seezoon.infrastructure.tcp.handler;

import com.seezoon.infrastructure.exception.Assertion;
import com.seezoon.infrastructure.tcp.codec.JT808;
import com.seezoon.infrastructure.tcp.codec.JTMessage;
import com.seezoon.infrastructure.tcp.codec.JTSerialization;
import com.seezoon.infrastructure.tcp.common.RpcContext;
import com.seezoon.infrastructure.tcp.jt808.JT0002;
import com.seezoon.infrastructure.tcp.jt808.JT8001;
import com.seezoon.infrastructure.tcp.session.ClientSession;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;

/**
 * 也可以响应读写事件 ，使用ChannelDuplexHandler
 */
@Slf4j
@Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<JTMessage> {

    private final static FutureManager futureManager = new FutureManager();
    private final DispatcherHandler dispatcher;

    public NettyClientHandler(DispatcherHandler dispatcher) {
        Assert.notNull(dispatcher, "dispatcher must not be null");
        this.dispatcher = dispatcher;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.debug("channelActive channel:{}", ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.debug("channelInactive channel:{}", ctx.channel());
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, JTMessage msg) throws Exception {
        ClientSession session = this.createSessionIfAbsent(msg.getSerialNo(), ctx);
        MessageHandler messageHandler = dispatcher.dispatch(msg.getMsgId());
        // 收服务端请求包
        if (messageHandler != null) {
            Object request = JTSerialization.decode(msg.getBody(), messageHandler.getRequestType());
            if (log.isDebugEnabled()) {
                log.debug("client channelRead request:{}, channel:{},msg:{} ", request, ctx.channel(), msg);
            }
            RpcContext rpcContext = creatRpcContext(msg);
            Object response = messageHandler.execute(request, rpcContext);
            if (log.isDebugEnabled()) {
                log.debug("client channelWrite response:{},context:{},msg:{},channel:{}", response, rpcContext, msg,
                        ctx.channel());
            }
            ctx.writeAndFlush(
                    createWriteMessage(messageHandler.responseType(), session.nextSequence(), rpcContext.getSerialNo(),
                            response));
        } else { // 收服务端响应
            // 应该消息2个字节为消息流水号
            if (msg.getBody().readableBytes() < 2) {
                log.error("client channelRead responseMsgSerialNo error length < 2 , channel:{},jtmessage:{}",
                        ctx.channel(), msg);
                return;
            }
            int responseMsgSerialNo = msg.getBody().getUnsignedShort(0);
            ResponseFuture responseFuture = futureManager.remove(ctx.channel().id().asLongText(), responseMsgSerialNo);
            if (responseFuture == null) { // 没有找到消息
                log.error("client channelRead no mapping responseMsgSerialNo:{},channel:{},jtmessage:{}",
                        responseMsgSerialNo, ctx.channel(), msg);
                return;
            }

            Object response = JTSerialization.decode(msg.getBody(), responseFuture.getClazz());
            if (log.isDebugEnabled()) {
                log.debug("client channelRead response:{} ,jtmessage:{},channel:{}", response, msg, ctx.channel());
            }
            responseFuture.getFuture().complete(response);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // server will close channel when server don't receive any heartbeat from client util timeout.
        if (evt instanceof IdleStateEvent) {
            // 发送心跳
            Channel channel = ctx.channel();
            ClientSession session = getSession(ctx);
            if (session == null) {
                log.debug("IdleStateEvent triggered no session close channel:{}", channel);
                ctx.close();
                return;
            }
            JTMessage writeMessage = this.createWriteMessage(JT808.终端心跳, session.nextSequence(),
                    session.getClientId(), new JT0002());
            int timeout = 60000;
            CompletableFuture<JT8001> future = this.responseFuture(channel.id().asLongText(),
                    writeMessage.getMsgSerialNo(), JT8001.class, timeout);
            ctx.writeAndFlush(writeMessage);
            // 使用异步方式处理心跳响应，避免阻塞事件循环
            future.thenAccept(jt8001 -> {
                log.debug("client channel:{} heartbeat response received: {}", channel, jt8001);
            }).exceptionally(throwable -> {
                log.error("client read heartbeat response error,close chanel:{}", channel, throwable);
                ctx.close();
                return null;
            });
            log.debug("IdleStateEvent triggered, send heartbeat, channel:{}", channel);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("client exceptionCaught context:{}", ctx, cause);
        ctx.close();
    }

    private RpcContext creatRpcContext(JTMessage msg) {
        RpcContext rpcContext = new RpcContext();
        rpcContext.setMsgId(msg.getMsgId());
        rpcContext.setSerialNo(msg.getSerialNo());
        rpcContext.setMsgSerialNo(msg.getMsgSerialNo());
        rpcContext.setDataLength(msg.getBodyLength());
        rpcContext.setPackageTotal(msg.getPackageTotal());
        rpcContext.setPackageNo(msg.getPackageNo());
        return rpcContext;
    }

    /**
     * @param msgId 回包消息ID
     * @param msgSerialNo 回包消息序号
     * @param serialNo 终端编号
     * @param response
     * @return
     */
    private JTMessage createWriteMessage(int msgId, int msgSerialNo, String serialNo, Object response) {
        byte[] encoded = JTSerialization.encode(response);
        // 服务单回包不考虑分包和加密
        JTMessage out = new JTMessage();
        out.setMsgId(msgId);
        out.setMsgProperties(encoded.length);
        out.setSerialNo(serialNo);
        out.setMsgSerialNo(msgSerialNo);
        out.setBody(Unpooled.wrappedBuffer(encoded));
        return out;
    }

    public <T> CompletableFuture<T> responseFuture(String channelId, long msgSerialNo, Class<T> clazz, long timeout) {
        Assertion.notEmpty(channelId, "add channelId not empty");
        Assertion.notNull(clazz, "add responseFuture clazz is not null");
        CompletableFuture<T> future = new CompletableFuture<>();
        futureManager.add(channelId, msgSerialNo, new ResponseFuture(clazz, future, timeout));
        return future;
    }

    private ClientSession getSession(ChannelHandlerContext ctx) {
        ClientSession session = ctx.channel().attr(ClientSession.key).get();
        return session;
    }

    private ClientSession createSessionIfAbsent(String serialNo, ChannelHandlerContext ctx) {
        Assertion.notEmpty(serialNo, "serialNo must not be empty");
        Assertion.notNull(ctx, "ctx must not be null");
        ClientSession session = ctx.channel().attr(ClientSession.key).get();
        if (session == null) {
            session = new ClientSession(serialNo);
        }
        ctx.channel().attr(ClientSession.key).set(session);
        return session;
    }
}
